using System;
using System.Linq;
using System.Reactive.Linq;
using RxHeat.Tracing;

namespace RxHeat.Threading
{
    /// <summary>
    ///     RxHeat Threading Observable helper extension methods
    /// </summary>
    public static class ObservableExtensions
    {
        /// <summary>
        ///     A replacement for the older Observable.ForkJoin. It will merge and wait for one response from all Observables
        /// </summary>
        /// <typeparam name="TSource">The type of the source.</typeparam>
        /// <param name="sources">The sources.</param>
        /// <param name="tracingKey">A tracing key to use during tracing exceptions. A default will be used when not supplied</param>
        /// <returns></returns>
        public static IObservable<AysncResult<TSource>[]> ForkJoinRx1<TSource>(string tracingKey = "ForkJoinRx1",
                                                                               params IObservable<TSource>[]
                                                                                   sources)
        {
            return Observable.Create<AysncResult<TSource>[]>(
                observer =>
                    {
                        var results = new AysncResult<TSource>[sources.Length];
                        IDisposable internalDisposable = null;
                        internalDisposable =
                            sources.Select((o, i) =>
                                           o.Select(x => new AysncResult<TSource>(x))
                                            .Catch((Exception ex) =>
                                                {
                                                    internalDisposable.TraceOnError(tracingKey, ex);
                                                    return Observable.Return(new AysncResult<TSource>(ex));
                                                })
                                            .TakeLast(1)
                                            .Do(v => results[i] = v)).Merge()
                                   .Count()
                                   .Where(c => c == sources.Length)
                                   .Select(i => results)
                                   .Subscribe(observer);
                        return internalDisposable;
                    });
        }
    }
}